package com.specialtreasure.freightdepot.service;

import android.content.Intent;
import android.os.AsyncTask;
import android.os.IBinder;
import android.support.annotation.Nullable;

import com.app.core.config.Constants;
import com.app.core.util.UserUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.specialtreasure.freightdepot.model.EventBean;
import com.specialtreasure.freightdepot.mq.MQMsg;
import com.specialtreasure.freightdepot.mq.MQueue;
import com.yanzhenjie.nohttp.Logger;

import org.simple.eventbus.EventBus;
import org.simple.eventbus.Subscriber;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by KUMA on 2017/8/16.
 */

public class MQService extends BaseService {
    public static final String INTENT_QUEUE = "INTENT_QUEUE";
    private Connection connection;
    private Channel channel;
    private Map<String, AsyncTask<MQueue, Void, Void>> taskMap = new HashMap<>();

    @Nullable
    @Override
    public IBinder onBind(Intent intent) {
        return null;
    }

    @Override
    public void onCreate() {
        super.onCreate();
        EventBus.getDefault().register(this);
        new AsyncTask<Void, Void, Void>() {
            @Override
            protected Void doInBackground(Void... voids) {
                try {
                    Logger.e("connection开始");
                    ConnectionFactory factory = new ConnectionFactory();
                    factory.setHost(Constants.MQ.MQ_IP);
                    factory.setPort(Constants.MQ.MQ_PORT);
                    factory.setUsername(Constants.MQ.MQ_USERNAME);
                    factory.setPassword(Constants.MQ.MQ_PASSWORD);
                    factory.setVirtualHost(Constants.MQ.MQ_VH);
                    factory.setAutomaticRecoveryEnabled(true);
                    //使用之前的设置，建立连接
                    connection = factory.newConnection();
                    //创建一个通道
                    channel = connection.createChannel();
                    //一次只发送一个，处理完成一个再获取下一个
                    channel.basicQos(1);
                    Logger.e("connection成功");


                    createQueue(Constants.MQ.ANDROIDONE+UserUtils.getUserId()
                            , Constants.MQ.GOODSADD + UserUtils.getUserId() + ".#"
                            , new OnReceiveMQListener() {
                                @Override
                                public void onReceiveMQ(String msg) {
                                    MQMsg mq = new MQMsg(msg);
                                    EventBus.getDefault().post(mq, Constants.MQ.ANDROIDONE);
                                }
                            }
                    );
                    createQueue(Constants.MQ.ANDROIDTWO+UserUtils.getUserId()
                            , Constants.MQ.GOODSDEL + UserUtils.getUserId() + ".#"
                            , new OnReceiveMQListener() {
                                @Override
                                public void onReceiveMQ(String msg) {
                                    MQMsg mq = new MQMsg(msg);
                                    EventBus.getDefault().post(mq, Constants.MQ.ANDROIDTWO);
                                }
                            }
                    );




                } catch (Exception e1) {
                    e1.printStackTrace();
                    Logger.e("connection失败");
                }
                return null;
            }

            @Override
            protected void onPostExecute(Void aVoid) {
                super.onPostExecute(aVoid);
            }
        }.execute();
    }

    @Override
    public int onStartCommand(Intent intent, int flags, int startId) {
        if (intent != null) {
            Object o = intent.getSerializableExtra(INTENT_QUEUE);
            if (o != null) {
                final MQueue q = (MQueue) o;
                startTask(q);
            }
        }
        return super.onStartCommand(intent, flags, startId);
    }

    //删除旧队列，创建新队列
    private void startTask(final MQueue q) {
        AsyncTask<MQueue, Void, Void> task = new AsyncTask<MQueue, Void, Void>() {
            @Override
            protected Void doInBackground(final MQueue... qs) {
                while (true) {
                    if (isCancelled()) {
                        return null;
                    }
                    if (channel != null && channel.isOpen()) {
                        break;
                    }
                    try {
                        Logger.e("channel尚未初始化，2秒后重试" + qs[0].queueName);
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                delQueue(qs[0].queueName);
                createQueue(qs[0].queueName, qs[0].routingKeys, new OnReceiveMQListener() {
                    @Override
                    public void onReceiveMQ(String msg) {
                        MQMsg mq = new MQMsg(msg);
                        EventBus.getDefault().post(mq, qs[0].eventTag);
                    }
                });
                return null;
            }

            @Override
            protected void onPostExecute(Void aVoid) {
                super.onPostExecute(aVoid);
                taskMap.remove(q.queueName);
            }
        };
        taskMap.put(q.queueName, task);
        task.execute(q);
    }

    private void createQueue(final String queueName, List<String> routingKeys, final OnReceiveMQListener onReceiveMQListener) {
        try {
            channel.queueDeclare(queueName, false, false, true, null);
            for (String routingKey : routingKeys) {
                channel.queueBind(queueName, Constants.MQ.MQ_EX, routingKey);
            }
            channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String message = new String(body);
                    Logger.e("_mq" + queueName + "_" + message);
                    onReceiveMQListener.onReceiveMQ(message);
                }
            });
            Logger.e(queueName + " 创建完毕");
        } catch (IOException e) {
            Logger.e(queueName + " 创建失败,e:" + e.getMessage());
            e.printStackTrace();
        }

    }

    private void createQueue(final String queueName, String routingKey, final OnReceiveMQListener onReceiveMQListener) {
        List<String> temp = new ArrayList<>();
        temp.add(routingKey);
        createQueue(queueName, temp, onReceiveMQListener);
    }

    public interface OnReceiveMQListener {
        public void onReceiveMQ(String msg);
    }

    @Override
    public void onDestroy() {
        EventBus.getDefault().unregister(this);
        if (connection != null) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        connection.close();
                        Logger.e("_mq" + "connection销毁");
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
        super.onDestroy();
    }

    private void delQueue(String queueName) {
        try {
            AsyncTask task = taskMap.get(queueName);
            if (task != null) {
                task.cancel(true);
                task = null;
            }
            taskMap.remove(queueName);
            channel.queueDelete(queueName);
            Logger.e("_mq" + queueName + "删除完毕");
        } catch (IOException e) {
            Logger.e("_mq" + queueName + "删除异常，e:" + e.getMessage());
            e.printStackTrace();
        }
    }




    @Subscriber(tag = "MQService")
    private void eventHandler(final EventBean eb) {
        switch (eb.what) {
            case Constants.MsgCode.ON_DELQUEUE:
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        delQueue((String) eb.getObj());
                    }
                }).start();
                break;
        }
    }

}
